package com.github.util;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.github.bean.ServerUser;
import com.github.manager.Manager;
import com.github.pojo.WebSocketMassage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

public class SocketSendThread {
    // 在控制器或服务中添加测试
    private static final Logger logger = LoggerFactory.getLogger(SocketSendThread.class);

    // 用于存储每个 WebSocketSession 对应的消息队列
    public static final BlockingQueue<WebSocketMassage> sessionQueues = new LinkedBlockingQueue<>();

    // 创建ThreadPoolExecutor线程池
    private static final ThreadPoolExecutor messageProcessorPool = new ThreadPoolExecutor(
            // 核心线程数
            Runtime.getRuntime().availableProcessors(),
            // 最大线程数
            Runtime.getRuntime().availableProcessors() * 2,
            // 空闲线程存活时间
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(),
            // 线程工厂
            r -> {
                Thread thread = new Thread(r);
                thread.setName("WebSocketMsgPool-" + thread.getId());
                thread.setDaemon(true);
                return thread;
            },
            new ThreadPoolExecutor.CallerRunsPolicy()
    );
    static {
        startMessageProcessors();
    }
    public static void startMessageProcessors() {
        new Thread(() -> {
            List<WebSocketMassage> batchMessages = new ArrayList<>();
            while (true) {
                try {
                    // 先尝试批量获取
                    // 先等待第一条消息（阻塞）
                    WebSocketMassage firstMessage = sessionQueues.take();
                    batchMessages.clear();
                    // 然后批量获取剩余消息
                    batchMessages.add(firstMessage);
                    sessionQueues.drainTo(batchMessages, 99);
                    // 处理批量消息
                    processBatchMessages(batchMessages);
                } catch (InterruptedException e) {

                }
            }
        }).start();
    }
    private static void processBatchMessages(List<WebSocketMassage> messages) {
        messages.forEach(msg ->
                messageProcessorPool.execute(() -> processSingleMessage(msg))
        );
    }

    private static void processSingleMessage(WebSocketMassage webSocketMassage) {
        WebSocketSession session = webSocketMassage.getSession();
        String sessionId = session.getId();

        try {
            if (session.isOpen()) {
                synchronized (session) {
                    if (session.isOpen()) {
                        String encryptedMessage = AESEncrypt.encrypt(webSocketMassage.getMessage());
                        session.sendMessage(new TextMessage(encryptedMessage));
                        logger.info("发送消息成功 sessionId{},msg{}", sessionId, webSocketMassage.getMessage());
                    }
                }
            }
        } catch (IOException e) {
            logger.warn("发送单条消息失败 session: {}", sessionId, e);
        }
    }


    public static void sendMessage(WebSocketSession session, String msg) {
        try {
            WebSocketMassage webSocketMassage = new WebSocketMassage();
            webSocketMassage.setSession(session);
            webSocketMassage.setMessage(msg);
            sessionQueues.put(webSocketMassage);
        }catch (Exception e) {
            logger.info("消息放入队列失败 sessionId:{},msg:{}",session.getId(),msg);
        }
    }
}
